package com.smsc.headend.task.engine.handler.impl.collection.strategy;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.google.common.base.Preconditions;
import com.smsc.headend.common.constant.Constant;
import com.smsc.headend.common.utils.CollectionIntervalsUtils;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.data.dto.DataItemDTO;
import com.smsc.headend.module.data.dto.TaskDataDTO;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.meterscheme.entity.CollMeterTaskPlan;
import com.smsc.headend.module.meterscheme.entity.LoadProfileGroupEntity;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.AtomicTaskState;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.dto.TaskReadData;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.service.CollMeterTaskPlanService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.collection.CollectTaskService;
import com.smsc.headend.task.engine.service.collection.RecollectTaskService;
import com.smsc.headend.task.engine.utils.TaskParamUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class ScheduleCollectionStrategy implements CollectionStrategy {
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    CollMeterTaskPlanService collMeterTaskPlanService;
    @Autowired
    AssetService assetService;
    @Autowired
    RecollectTaskService recollectTaskService;
    @Autowired
    CollectTaskService collectTaskService;

    @Override
    public List<RdMeterReadsLp> filterCollectionData(Task task, AtomicTask atomicTask, Map<String, Object> param, List<RdMeterReadsLp> meterReadsCol) {
        return meterReadsCol;
    }

    @Override
    public int callback(Meter meter, TaskReadData readData, Task task, AtomicTask atomicTask, Map<String, Object> param, List<DataItemDTO> dataItemCosemList, List<RdMeterReadsLp> filteredRdMeterReadsLp) {
        AtomicTaskState atomicTaskState = taskDAO.hashGetAtomicTaskState(atomicTask.getAtomicTaskNo());
        boolean isAnsiProfile = Optional.ofNullable(atomicTaskState.getData()).orElse(Collections.EMPTY_MAP).containsKey("CURRENT_INTERVAL_OFFSET");
        boolean isCursorArriveEndPoint = isAnsiProfile ? true : MapUtil.getLong(param, "profileEndTime", 0L).equals(atomicTaskState.getCollectingEndTv());
        boolean snapshot = false;
        if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            String profileType = MapUtil.getStr(param, "profileType");
            if (StrUtil.isEmpty(profileType) && LoadProfileGroupEntity.ProfileType.Snapshot.getCode() == Integer.parseInt(profileType)) {
                snapshot = true;
            }
        }
        if(snapshot) {
            return 1;
        }
        if (CollectionUtils.isEmpty(filteredRdMeterReadsLp) && isCursorArriveEndPoint) {
            if (!isAnsiProfile) {
                updateProfileProgress(task, atomicTask, filteredRdMeterReadsLp, meter);
            }
            log.info("sch_read: read empty data; meterId={}, param={}", task.getMeterId(), JSONUtil.toJsonStr(param));
            return 1;
        }

        // ansi 默认是最后一个点， 不需要比对, cosem 要计算是否到结尾
        if (!isAnsiProfile) {
            Long taskFinalEndTv = Long.parseLong(String.valueOf(param.get("profileEndTime")));
            Long taskStartTv = Long.parseLong(String.valueOf(param.get("profileStartTime")));
            Long interval = Convert.toLong(param.get("interval"));
            String recollCreate = assetService.getSystemProperty("RECOLL_CREATE");
            boolean recollectOn = Constant.TRUE.equalsIgnoreCase(recollCreate);
            if (recollectOn && interval != null) {
                handleRecollectPointInfo(task, interval, taskStartTv, taskFinalEndTv, atomicTaskState, filteredRdMeterReadsLp, isCursorArriveEndPoint);
            }

            if (!isCursorArriveEndPoint) {
                log.debug("sch_read: collecting next time span, meterId:{},target:{},complete:{}", task.getMeterId(), atomicTaskState.getCollectingEndTv(), atomicTaskState.getCollectedEndTv());
                atomicTaskState.setCollectedEndTv(atomicTaskState.getCollectingEndTv());
                atomicTaskState.setStored(false);
                taskDAO.hashSetAtomicTaskState(atomicTask.getAtomicTaskNo(), task, atomicTaskState.getRetryTimes(), atomicTaskState);
                return 0;
            }
        }
        updateProfileProgress(task, atomicTask, filteredRdMeterReadsLp, meter);
        return 1;
    }

    private void updateProfileProgress(Task task, AtomicTask atomicTask, List<RdMeterReadsLp> filteredRdMeterReadsLp, Meter meter) {
        Long progressTime;
        if (CollectionUtils.isEmpty(filteredRdMeterReadsLp)) {
            TaskDataDTO latestTv = Optional.ofNullable(taskDAO.popAllTaskData(task.getTaskNo())).orElse(Collections.emptyList()).stream()
                    .filter(dto -> dto.getKey().equals("latestTv")).findAny().orElse(null);
            progressTime = latestTv == null ? null : Convert.toLong(latestTv.getValue());
        } else {
            progressTime = filteredRdMeterReadsLp.stream().max(Comparator.comparing(RdMeterReadsLp::getTv)).get().getTv();
        }

        if (progressTime != null) {
            Long dataItemId = MapUtil.getLong(TaskParamUtils.getParameterMap(task, atomicTask), "dataItemId");
            log.info("setMeterProgressUpdateMessage:meterId:{},lpId:{},tv:{}", task.getMeterId(), dataItemId, progressTime);
            CollMeterTaskPlan collMeterTaskPlan = collMeterTaskPlanService.getCollMeterTaskPlan(meter.getMeterId(), dataItemId);
            if (collMeterTaskPlan == null || collMeterTaskPlan.getProgressTime() < progressTime) {
                collMeterTaskPlanService.saveMeterCollectProgress(progressTime, task.getMeterId(), dataItemId);
            }
        }
    }

    private void handleRecollectPointInfo(Task task, Long interval, Long taskStartTv, Long taskFinalEndTv, AtomicTaskState atomicTaskState, List<RdMeterReadsLp> filteredRdMeterReadsLp, boolean isCursorArriveEndPoint) {
        Preconditions.checkArgument(filteredRdMeterReadsLp.size() > 0);
        Long startTv = atomicTaskState.getCollectedEndTv();
        Long endTv = atomicTaskState.getCollectingEndTv();
        String taskNo = task.getTaskNo();
        Long meterId = task.getMeterId();
        Long lpDataItemId = filteredRdMeterReadsLp.get(0).getLpDataItemId();
        Set<Long> collectTv = filteredRdMeterReadsLp.stream().map(RdMeterReadsLp::getTv).collect(Collectors.toSet());
        Set<Long> absentPointSet = null;
        //read profile的时间未拆分
        if (taskStartTv.equals(startTv) && taskFinalEndTv.equals(endTv)) {
            absentPointSet = CollectionIntervalsUtils.getAbsentPointList(startTv, endTv, collectTv, interval);
        } else {
            //拆分了请求的定抄任务，暂存tv值，在完成时取出所有tv并计算缺点
            if (isCursorArriveEndPoint) {
                Set<Long> collectTvs = collectTaskService.getAndInvalidLpTv(taskNo);
                absentPointSet = CollectionIntervalsUtils.getAbsentPointList(startTv, endTv, collectTvs, interval);
            } else {
                collectTaskService.saveLpTv(taskNo, collectTv);
            }
        }

        if (!CollectionUtils.isEmpty(absentPointSet)) {
            log.info("recollect: absent point,meterId:{},lpId:{},tv:{}", meterId, lpDataItemId, JSONUtil.toJsonStr(absentPointSet));
            recollectTaskService.checkRecollectTask(meterId, interval, absentPointSet, lpDataItemId);
        }
    }
}
